package com.sparrow.common.schedule.playerscheduler;

import com.sparrow.common.schedule.playerscheduler.listener.PlayerSchedulerListener;

import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.function.Consumer;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

/**
 * 玩家调度器
 * 1最初设计是为了避免player.schedule(p -> p.doSomething());调度的时候实质上所有玩家共享ScheduledExecutorService 去缓存所有任务， 玩家下线后 lambada 闭包包含了Player强引用，玩家下线没有即使释放的问题
 * 2避免了玩家所有延迟任务都丢到一个共享的调度器里面
 */
public class PlayerScheduler<E> {
    private ScheduledExecutorService executor;
    private final Queue<ScheduledTask<E>> taskQueue = new PriorityQueue<>();
    private volatile PlayerScheduledFuture currentFuture;
    private volatile boolean stop;
    private CopyOnWriteArrayList<PlayerSchedulerListener<E>> listeners = new CopyOnWriteArrayList<>();
    private E e;

    public PlayerScheduler(ScheduledExecutorService executor, E owner) {
        this.executor = executor;
        this.e = owner;
    }

    public void addListener(PlayerSchedulerListener listener) {
        listeners.add(listener);
    }

    public void execute(Consumer<E> task) {
        if (stop) {
            return;
        }
        executor.execute(() -> {
            task.accept(e);
        });
    }

    public synchronized void schedule(String taskName, Consumer<E> task, long delay, TimeUnit unit) {
        schedule(taskName, task, unit.toMillis(delay), 0);
    }

    public synchronized void scheduleAtFixedRate(String taskName, Consumer<E> task, long initialDelay, long delay, TimeUnit unit) {
        if (delay <= 0) {
            throw new IllegalArgumentException("intervalMillis must be > 0");
        }
        schedule(taskName, task, unit.toMillis(initialDelay), unit.toMillis(delay));
    }

    public synchronized void scheduleWithFixedDelay(String taskName, Consumer<E> task, long initialDelay, long delay, TimeUnit unit) {
        if (delay <= 0) {
            throw new IllegalArgumentException("intervalMillis must be > 0");
        }
        schedule(taskName, task, unit.toMillis(initialDelay), -unit.toMillis(delay));
    }

    private void schedule(String taskName, Consumer<E> task, long initialDelay, long delay) {
        if (stop) {
            return;
        }
        ScheduledTask scheduledTask = new ScheduledTask(taskName, task, System.currentTimeMillis() + initialDelay, delay);
        addTask(scheduledTask);
    }

    private synchronized void addTask(ScheduledTask scheduledTask) {
        taskQueue.offer(scheduledTask);
        // 只有新任务更早才打断重排
        if (currentFuture == null || scheduledTask.executeTime < currentFuture.task.executeTime) {
            cancelCurrentTask();
            scheduleNext();
        }
    }

    private void cancelCurrentTask() {
        if (currentFuture != null) {
            currentFuture.cancel();
            currentFuture = null;
        }
    }

    public void reset() {
        stop = false;
    }

    public synchronized void shutdown() {
        stop = true;
        cancelCurrentTask();
        taskQueue.clear();

        for (PlayerSchedulerListener<E> listener : this.listeners) {
            listener.onPlayerSchedulerShutdown(e);
        }
    }

    private synchronized void scheduleNext() {
        ScheduledTask next = taskQueue.peek();
        if (next == null) {
            return;
        }
        if (stop) {
            return;
        }
        long now = System.currentTimeMillis();
        long delay = Math.max(0, next.executeTime - now);
        ScheduledFuture<?> f = executor.schedule(() -> {
            executeTask(next);
        }, delay, MILLISECONDS);
        this.currentFuture = new PlayerScheduledFuture(f, next);
    }

    private void executeTask(ScheduledTask expected) {
        ScheduledTask decorateTask;
        synchronized (this) {
            if (stop) {
                return;
            }
            decorateTask = taskQueue.poll();
            if (decorateTask == null || decorateTask != expected) {
                System.err.println(String.format("expect task vs poll task not same expected=%s poll=%s", expected.getTaskName(), decorateTask.getTaskName()));
                if (decorateTask != null) {
                    taskQueue.offer(decorateTask);
                }
                scheduleNext();
                return;
            }
        }
        try {
            decorateTask.task.accept(e);
        } catch (Exception ex) {
            System.err.println("Error in scheduled task: " + decorateTask.taskName);
            ex.printStackTrace();
        }

        onTaskFinished(decorateTask);
    }

    private synchronized void onTaskFinished(ScheduledTask task) {
        currentFuture = null;
        if (stop) {
            return;
        }
        if (task.isPeriod()) {//循环执行的任务需要从新放到任务队列
            if (task.getPeriod() > 0) { //固定频率
                task.setExecuteTime(task.executeTime + task.period);
            } else { //固定延迟 period 是个负数
                task.setExecuteTime(System.currentTimeMillis() - task.period);
            }
            addTask(task);
        } else {
            scheduleNext();
        }
    }

    private static class ScheduledTask<E> implements Comparable<ScheduledTask<E>> {
        final Consumer<E> task;
        long executeTime; // 绝对执行时间戳 ms
        /**
         * 沿用jdk ScheduledThreadPoolExecutor period设计
         * 0 表示一次性
         * > 0 表示固定频率 scheduleAtFixedRate
         * < 0 固定延迟 scheduleWithFixedDelay
         */
        final long period;
        final String taskName;

        ScheduledTask(String taskName, Consumer<E> task, long executeTime, long period) {
            this.taskName = taskName;
            this.task = task;
            this.executeTime = executeTime;
            this.period = period;
        }

        public String getTaskName() {
            return taskName;
        }

        public boolean isPeriod() {
            return period != 0;
        }

        public long getPeriod() {
            return period;
        }

        public long getExecuteTime() {
            return executeTime;
        }

        public int compareTo(ScheduledTask<E> other) {
            return Long.compare(this.executeTime, other.executeTime);
        }

        public void setExecuteTime(long executeTime) {
            this.executeTime = executeTime;
        }
    }

    public static class PlayerScheduledFuture {
        final ScheduledFuture future;
        final ScheduledTask task;

        public PlayerScheduledFuture(ScheduledFuture future, ScheduledTask task) {
            this.future = future;
            this.task = task;
        }

        public void cancel() {
            this.future.cancel(false);
        }

    }

}
